﻿using System;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Threading;
using System.Linq;
using System.Net;

using RexToy.Logging;
using RexToy.DesignPattern;


namespace RexToy.Cassandra
{
    public sealed class Cluster : DisposableBase
    {
        private static ILog _log = LogContext.GetLogger<Cluster>();

        private ILoadBalancePolicy _lbPolicy;
        private IPoolingPolicy _poolingPolicy;
        private int _poolSizeHint;
        private int _timeOut;
        public Cluster(LoadBalance loadBalance = LoadBalance.RoundRobin, Pooling pooling = Pooling.ExhaustedBlock, int poolSizeHint = 20, int timeOut = 30000)
        {
            _lbPolicy = ClusterPolicyFactory.CreateLoadBalancePolicy(loadBalance);
            _poolingPolicy = ClusterPolicyFactory.CreatePoolingPolicy(pooling);
            Assertion.IsTrue(poolSizeHint > 0, "Pool size [{0}] not correct.", poolSizeHint);
            _poolSizeHint = poolSizeHint;
            _timeOut = timeOut;
        }

        public Cluster(int poolSizeHint, int timeOut = 30000)
            : this(LoadBalance.RoundRobin, Pooling.ExhaustedBlock, poolSizeHint, timeOut)
        {
        }

        private bool _initialized = false;
        private List<DnsEndPoint> _hosts;
        private Dictionary<DnsEndPoint, ConcurrentQueue<RawConnection>> _dict;
        public void Initialize(List<DnsEndPoint> hosts)
        {
            hosts.ThrowIfNullArgument("hosts");
            Assertion.IsTrue(hosts.Count > 0, "Host list is empty.");
            Assertion.AreEqual(hosts.Count, hosts.Distinct().Count(), "Host list have duplicate entry.");
            Assertion.IsFalse(_initialized, "Cluster already initialized.");

            _hosts = hosts;
            int sizePerHost = _poolSizeHint / hosts.Count + 1;
            _dict = new Dictionary<DnsEndPoint, ConcurrentQueue<RawConnection>>();
            foreach (var ep in hosts)
            {
                var q = new ConcurrentQueue<RawConnection>();
                for (int i = 0; i < sizePerHost; i++)
                {
                    var rc = new RawConnection(ep);
                    q.Enqueue(rc);
                }

                _dict[ep] = q;
            }

            _lbPolicy.Initialize(hosts);
            _poolingPolicy.Initialize(hosts, sizePerHost * hosts.Count, _timeOut);
            _initialized = true;
        }

        public void Initialize(params DnsEndPoint[] hosts)
        {
            List<DnsEndPoint> list = new List<DnsEndPoint>(hosts);
            Initialize(list);
        }

        public int AvailableResource
        {
            get
            {
                CheckDisposed();
                return _poolingPolicy.AvailableResource;
            }
        }

        internal RawConnection GetRawConnection()
        {
            CheckDisposed();

            var ctx = new Context();
            _poolingPolicy.Pick(ctx);
            _lbPolicy.Pick(ctx);

            if (ctx.Stop)
                throw ConnectExceptionHelper.CreateConnectError(ctx);

            var ep = _hosts[ctx.SelectedIndex];
            IProducerConsumerCollection<RawConnection> q = _dict[ep];

            RawConnection cnn;
            bool taken = false;
            do
            {
                taken = q.TryTake(out cnn);
            } while (!taken);

            return cnn;
        }

        internal void PutRawConnection(RawConnection cnn)
        {
            CheckDisposed();

            var ctx = new Context();
            ctx.SelectedIndex = _hosts.IndexOf(cnn.EndPoint);
            _lbPolicy.Recycle(ctx);
            _poolingPolicy.Recycle(ctx);

            var q = _dict[cnn.EndPoint];
            if (cnn.IsOpen)
                cnn.Close();
            q.Enqueue(cnn);
        }

        protected override void DisposeManagedObject()
        {
            _poolingPolicy.Dispose();
            foreach (var kvp in _dict)
            {
                var q = kvp.Value;
                RawConnection cnn;
                while (q.TryDequeue(out cnn))
                    cnn.Dispose();
            }
        }

        internal class Context
        {
            public int SelectedIndex { get; set; }
            public bool Stop { get; set; }
            public StopReason Reason { get; set; }
        }
    }
}
